#include <unistd.h>
#include "service.h"
#include "z_logger.h"
#include "aiot_state_api.h"
#include "aliot.h"
#include "../mqtt/mqtt.h"
#include "core_mqtt.h"
#include <sys/stat.h>
#include <sys/inotify.h>

#define TAG_HEAD1 0b1010101 //二进制55
#define TAG_HEAD2 0b10101010 //二进制aa
#define EVENT_BUF_LEN ( 1024 * ( EVENT_SIZE + 16 ) )
#define EVENT_SIZE  ( sizeof (struct inotify_event) )


static block_t block;
static long read_position;//读取位置
static char topic[NAME_MAX];
//hisdata存放文件名及路径
static char file_name[NAME_MAX];
static char log_file[NAME_MAX] = "/usr/data/loaded.db";

//根据loaded.db文件选择读取位置
static int32_t find_position() {
    int32_t log_position;//记录位置
    FILE *file = fopen(log_file, "rb");
    if (NULL == file) {
        //第一次未记录，直接读取hdb
        log_position = 0;
    } else {
        //从最后一个记录数据包的位置读取一个
        long size = sizeof(uint32_t);
        fseek(file, -size * 2, SEEK_END);
        uint32_t t;
        fread(&t, sizeof(uint32_t), 1, file);
        time_t log = t;
        time_t now = time(NULL);
        struct tm *log_time, *now_time;
        now_time = localtime(&now);
        int now_day = now_time->tm_mday;
        log_time = localtime(&log);
        int log_day = log_time->tm_mday;
        //比较loaded.db中的timestamp时间戳和当前时间的日是否相同，不同则为新的一天
        if (now_day != log_day) {
            return 0;
        }
        fread(&log_position, sizeof(uint32_t), 1, file);
        fclose(file);
    }
    return log_position;
}

static int payload_send(void *handle, uint16_t dev_id, block_t *block_send, uint32_t *timestamp,
                        tag_header_t *tag_send, float *value) {
    int rc;
    rc = publish_scada_aliot(value, topic, handle, dev_id, block_send, timestamp, tag_send);
    if (rc < 0)
        log_error("aliot pub failed!");
    return rc;
}

//读取数据
static int read_payload_pub(void *handle, char *file_path) {
    int rc, rs = 0;
    FILE *file = fopen(file_path, "rb");
    //定位至上次发送后的位置
    fseek(file, find_position(), SEEK_SET);
    read_position = ftell(file);
    log_trace("file position: %ld", read_position);

    //读取头部
    while (0 != fread(&block.head, sizeof(block.head), 1, file)) {
        uint16_t data_length = block.head.len;

        char data_buf[data_length];
        bzero(data_buf, sizeof(data_buf));

        //读取数据部分
        fread(data_buf, data_length, 1, file);

        //读取尾部
        fread(&block.tail, sizeof(block.tail), 1, file);

        read_position = ftell(file);

        //判断存储是否有问题
        if (TAG_HEAD1 != block.tail.eof[1] && TAG_HEAD2 != block.tail.eof[0]) {
            log_error("数据块尾部错误！");
            //轮询偏移值，去寻找最近的一个blockhead开头的
            int cycle = 1;
            fseek(file, find_position() + cycle, SEEK_SET);
            bzero(data_buf, sizeof(data_buf));
            while (0 != fread(&block.head, sizeof(block.head), 1, file)) {
                if (TAG_HEAD1 == block.head.bof[0] && TAG_HEAD2 == block.head.bof[1]) {
                    fseek(file, find_position() + cycle, SEEK_SET);
                    read_position = ftell(file);
                    break;
                }
                fseek(file, find_position() + ++cycle, SEEK_SET);
            }
        } else {
            void *p;
            uint32_t *timestamp = &block.head.timestamp;
            p = data_buf;
            hdb_header_t *hdb_header = (hdb_header_t *) p;
            int num_tag = hdb_header->n_tags;
            p += sizeof(hdb_header_t);
            //循环num_tag个tag_header和具体的tag_value
            for (int i = 0; i < num_tag; i++) {
                tag_header_t *tag_header = (tag_header_t *) p;
                p += sizeof(tag_header_t);
                //读取value长度
                float *value = p;
                p += sizeof(float);
                if (tag_header->has_timestamp) {
                    timestamp = p;
                    p += sizeof(uint32_t);
                }
                rc = payload_send(handle, hdb_header->dev_id, &block, timestamp, tag_header, value);
                if (rc < 0) {
                    rs = rc;
                    log_error("aliot_send error!");
                }
                //睡眠超过0.1s，保证传送阿里云不出现429流量限制，1s不超过10条信息
                log_trace("sleep 1s to wait!");
                usleep(1000);//101
            }
        }
    }
    fclose(file);
    return rs;
}

static void inotify_end(int fd, int wd) {
    inotify_rm_watch(fd, wd);
    close(fd);
}

static void write_read_postion(FILE *file) {
    struct timeval _tv;
    gettimeofday(&_tv, NULL);
    uint32_t timestamp = _tv.tv_sec;
    uint32_t p = read_position;
    log_trace("read_position=%ld", read_position);
    size_t result;
    result = fwrite(&timestamp, sizeof(timestamp), 1, file);
    if (result < 0) log_error("fwrite error!");
    result = fwrite(&p, sizeof(p), 1, file);
    if (result < 0) log_error("fwrite error!");
}

static int flush_write(FILE *file) {
    int rc;
    rc = fflush(file);
    if (0 != rc) log_error("fflush error!");
    return rc;
}

//合并获取完整文件路径
static void path_combine(char *file_path) {
    time_t cur_time = time(NULL);
    struct tm _tm;
    localtime_r(&cur_time, &_tm);
    sprintf(file_path, "/usr/data/");
    snprintf(file_name, sizeof(file_name), "hdb%04d%02d%02d.db",
             _tm.tm_year + 1900, _tm.tm_mon + 1, _tm.tm_mday);
    strcat(file_path, file_name);
}


//检查判断是否上传
static void pub_pull_checked(int fd, void *handle, char *file_path) {
    int rc;
    //文件更新，需要上传
    char buf[EVENT_BUF_LEN];
    ssize_t len = read(fd, buf, EVENT_BUF_LEN);
    int i = 0;
    if (len < 0) {
        log_error("read error!");
        goto
                end;
    }
    while (i < len) {
        struct inotify_event *evt = (struct inotify_event *) &buf[i];
        if (evt->mask & IN_CLOSE) {
            //代表该文件写入完成
            rc = read_payload_pub(handle, file_path);
            //上传成功，记录文件读取位置
            if (rc >= 0) {
                FILE *file = fopen(log_file, "a");
                write_read_postion(file);
                //从缓存区全部写入
                flush_write(file);
                if (NULL != file) fclose(file);
            }
        }
        i += EVENT_SIZE + evt->len;
    }
    end:
    log_trace("pub finished.");
}

static void *_pub_service_init(aliot_service_args_t *args) {
    uplink_t *uplink = args->seqiot->default_uplink;
    sprintf(topic, "/%s/%s/user/scada", uplink->product_key, uplink->device_key);
    //订阅scada主题
    aiot_mqtt_sub(args->mqtt_handle, topic, NULL, 1, NULL);
    return args->mqtt_handle;
}

static void *_pub_service_start(void *handle) {
    int32_t res = STATE_SUCCESS;

    if (!handle) {
        aliot_error_check("pub_service_start", STATE_USER_INPUT_NULL_POINTER);
        return NULL;
    }
    char file_path[PATH_MAX] = "/usr/data/";
    path_combine(file_path);
    int fd = inotify_init();
    int wd = inotify_add_watch(fd, file_path, IN_CLOSE);

    while (running) {
        path_combine(file_path);
        sleep(5);
        pub_pull_checked(fd, handle, file_path);
    }

    inotify_end(fd, wd);
    log_debug("pub_thread_exit");

    return NULL;
}

static aliot_service_t _pub_service = {
        .name = "pub",
        .init = _pub_service_init,
        .fini = NULL,
        .start = _pub_service_start,
        .stop = NULL,
};

aliot_service_t *pub_service_get() {
    return &_pub_service;
}